package com.chb.userbehavioranalysis.hotitem

import java.io.{BufferedReader, FileInputStream,InputStreamReader}
import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

object KafkaProducer {

    def main(args: Array[String]): Unit = {
        writeToKafka("hotitems")
    }

    def writeToKafka(topic: String): Unit = {
        val props = new Properties()
        props.put("bootstrap.servers", "10.0.0.201:9092")
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

        val producer = new KafkaProducer[String, String](props)
        val dataURL = getClass.getResource("/UserBehavior.csv")
        val br = new BufferedReader(new InputStreamReader(new FileInputStream(dataURL.getPath)))
        var line = br.readLine()
        while (line != null) {
            val record = new ProducerRecord[String, String](topic, line)
            producer.send(record)
            Thread.sleep(100)
            line = br.readLine()
        }
        producer.close()
    }
}